package com.yin.myproject.distributedlock;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;

import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author Eason
 * @create 2018-03-06 11:01
 **/
public class BaseDistributedLock {

    private static final Integer MAX_RETRY_COUNT = 10;
    private final ZkClient client;
    private final String path;
    //zookeeper中locker节点的路径
    private final String basePath;
    private final String lockName;

    public BaseDistributedLock(ZkClient client, String path, String lockName) {
        this.client = client;
        this.basePath = path;
        this.path = path.concat("/").concat(lockName);
        this.lockName = lockName;
    }

    private void deleteOurPath(String ourPath) throws Exception {
        client.delete(ourPath);
    }

    private String createLockNode(ZkClient client, String path) throws Exception {
        return client.createEphemeralSequential(path, null);
    }

    private boolean waitToLock(long startMillis, Long millisToWait, String ourPath) throws Exception {
        boolean haveTheLock = false;
        boolean doDelete = false;
        try {
            while (!haveTheLock) {
                //获取lock节点下的所有节点
                List<String> children = getSortedChildren();
                String sequenceNodeName = ourPath.substring(basePath.length() + 1);
                //获取当前节点的在所有节点类表中的位置
                int ourIndex = children.indexOf(sequenceNodeName);
                //节点位置小于0,说明没有找到节点
                if (ourIndex < 0) {
                    throw new ZkNoNodeException("节点没有找到:" + sequenceNodeName);
                }
                //节点位置大于0说明还有其他节点在当前节点前面,就需要等待其他的节点都释放
                boolean isGetTheLock = ourIndex == 0;
                String pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1);
                if (isGetTheLock) {
                    haveTheLock = true;
                } else {
                    //获取当前节点次小的节点,并监听节点的变化
                    String previousSequencePath = basePath.concat("/").concat(pathToWatch);
                    final CountDownLatch latch = new CountDownLatch(1);
                    final IZkDataListener priviousListiner = new IZkDataListener() {
                        @Override
                        public void handleDataChange(String s, Object o) throws Exception {
                            latch.countDown();
                        }

                        @Override
                        public void handleDataDeleted(String s) throws Exception {

                        }
                    };

                    try {
                        client.subscribeDataChanges(previousSequencePath, priviousListiner);
                        if (millisToWait != null) {
                            millisToWait -= (System.currentTimeMillis() - startMillis);
                            startMillis = System.currentTimeMillis();
                            if (millisToWait <= 0) {
                                doDelete = true;
                                break;
                            }
                            latch.await(millisToWait, TimeUnit.MICROSECONDS);
                        } else {
                            latch.await();
                        }
                    } finally {
                        client.unsubscribeDataChanges(previousSequencePath, priviousListiner);
                    }


                }
            }
        } catch (Exception e) {
            //发生异常需要删除节点
            doDelete = true;
            throw e;
        } finally {
            if (doDelete) {
                deleteOurPath(ourPath);
            }
        }

        return haveTheLock;
    }

    private List<String> getSortedChildren() {
        List<String> children = client.getChildren(basePath);
        Collections.sort(children, Comparator.comparing(o -> getLockNodeNumber(o, lockName)));
        return children;
    }

    private String getLockNodeNumber(String str, String lockName) {
        int index = str.lastIndexOf(lockName);
        if (index >= 0) {
            index += lockName.length();
            return index <= str.length() ? str.substring(index) : "";
        }
        return str;
    }

    protected String attemptLock(long time, TimeUnit unit) throws Exception {
        final long startMillis = System.currentTimeMillis();
        final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;
        String ourPath = null;
        boolean hasTheLock = false;
        boolean isDone = false;
        int retryCount = 0;
        while (!isDone) {
            isDone = true;
            try {
                ourPath = createLockNode(client, path);
                hasTheLock = waitToLock(startMillis, millisToWait, ourPath);
            } catch (ZkNoNodeException e) {
                if (retryCount++ < MAX_RETRY_COUNT) {
                    isDone = false;
                } else {
                    throw e;
                }
            }
        }
        if (hasTheLock) {
            return ourPath;
        }
        return null;
    }

    protected void releaseLock(String lockPath) throws Exception {
        deleteOurPath(lockPath);
    }
}
